package com.atguigu.gmall.app.dws.log;

import com.atguigu.gmall.bean.BeanTwo;
import com.atguigu.gmall.bean.TableBean;
import com.atguigu.gmall.utils.ClickHouseUtil;
import com.atguigu.gmall.utils.SQLUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        String topicName = "dwd_traffic_page_log";
        String groupID = "channel_flow_app";
        tEnv.executeSql("create table dwd_traffic_page_log( " +
                " `common` map<string,string>, " +
                " `page` map<string,string>, " +
                " `ts` bigint, " +
                " `rt` AS TO_TIMESTAMP_LTZ(ts,3), " +
                "  WATERMARK FOR rt AS rt - INTERVAL '0' SECOND " +
                ")" + SQLUtil.getKafkaSource(topicName,groupID));

        String topicName2 = "dwd_traffic_unique_visitor_detail";
        String groupID2 = "channel_flow_app";
        tEnv.executeSql("create table dwd_traffic_unique_visitor_detail( " +
                " `common` map<string,string>, " +
                " `page` map<string,string>, " +
                " `ts` bigint " +
                ")" + SQLUtil.getKafkaSource(topicName2,groupID2));

        Table table1 = tEnv.sqlQuery(
                " select  " +
                        "  substr(CAST(FROM_UNIXTIME(UNIX_TIMESTAMP()) AS string),0,13) h, " +
                " count(distinct common['is_new']) midCount " +
                " from dwd_traffic_unique_visitor_detail " +
                " ");

        tEnv.createTemporaryView("tyuut",table1);


        Table table2 = tEnv.sqlQuery(
                        " select " +
                        " stt, " +
                        " edt, " +
                        " yt.midCount midCount, " +
                        " pageIdCount, " +
                        " avgPageId, " +
                        " ui.h h, " +
                        " ts " +
                        "from( " +
                        " select " +
                        " DATE_FORMAT(TUMBLE_START(rt, interval '5' second), 'yyyy-MM-dd HH:mm:ss') stt, " +
                        " DATE_FORMAT(TUMBLE_END(rt, interval '5' second), 'yyyy-MM-dd HH:mm:ss')  edt," +
                        " count(page['page_id']) pageIdCount, " +
                        " count(page['page_id'])/count(distinct common['sid']) avgPageId," +
                        " substr(CAST(FROM_UNIXTIME(UNIX_TIMESTAMP()) AS string),0,13) h, " +
                        " unix_timestamp()*1000 ts " +
                        " from " +
                        " dwd_traffic_page_log " +
                        " group by substr(CAST(FROM_UNIXTIME(UNIX_TIMESTAMP()) AS string),0,13)," +
                        " TUMBLE(rt, INTERVAL '5' SECOND)) as ui " +
                        " left join tyuut yt on ui.h = yt.h  " +
                        " " );

        DataStream<Tuple2<Boolean, BeanTwo>> tuple2DataStream = tEnv.toRetractStream(table2, BeanTwo.class);
        SingleOutputStreamOperator<BeanTwo> tableBeanSingleOutputStreamOperator = tuple2DataStream.flatMap(new FlatMapFunction<Tuple2<Boolean, BeanTwo>, BeanTwo>() {
            @Override
            public void flatMap(Tuple2<Boolean, BeanTwo> value, Collector<BeanTwo> out) throws Exception {
                if (value.f0 = true) {
                    out.collect(value.f1);
                }
            }
        });

        tableBeanSingleOutputStreamOperator.print();
        String sql = "insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?,?)";
        tableBeanSingleOutputStreamOperator.addSink(ClickHouseUtil.<BeanTwo>getJdbcSink(sql));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
